//
// Created by 成飞 on 2024/3/5.
//
#include "server.h"
#include "meta_data.h"
#include "util/thread_pool.hpp"
#include "config/config_data.h"
#include <algorithm>
#include <unistd.h>
#include <fcntl.h>
#include <thread>
#include <sys/mman.h>
#include <sstream>
#include <iostream>
#include <filesystem>
#include <vector>
#include <sys/stat.h>

namespace db{
    namespace fs = std::filesystem;

    Server::Server() {
        conf::initConfData();
        meta::init();
        meta::load();
        meta::storeInfo();
        crash_check_gc();
        meta::initLog();
        //todo repair dataLog or gc
        //如果是readyToGC 192 cleanFileNo 786 192就说明完整的，崩溃时未gc，否则就要把786重新clean
        init();
        crash_check_data();
    }

    static double rangeQn = 0;
    static double tm12 = 0;
    static double tm23 = 0;
    static double tm34 = 0;
    static double tm45 = 0;

    static double scanTm12 = 0;
    static double scanTm23 = 0;
    static double scanTm34 = 0;
    static double scanTm45 = 0;

    Server::~Server(){
        meta::closeDB.store(true);
        meta::closeLog();
        meta::storeInfo();

        using namespace std;
        cout << "tm12: " << tm12 / rangeQn << " microseconds" << endl;
        cout << "tm23: " << tm23 / rangeQn << " microseconds" << endl;
        cout << "tm34: " << tm34 / rangeQn << " microseconds" << endl;
        cout << "tm45: " << tm45 / rangeQn << " microseconds" << endl;

        cout << "scanTm12: " << scanTm12 / rangeQn << " microseconds" << endl;
        cout << "scanTm23: " << scanTm23 / rangeQn << " microseconds" << endl;
        cout << "scanTm34: " << scanTm34 / rangeQn << " microseconds" << endl;
        cout << "scanTm45: " << scanTm45 / rangeQn << " microseconds" << endl;

        for (const auto &item: meta::fileControlInfo)
            fclose(item.second.fd);
        for (const auto &item: WBufferCluster){
//            delete item;
        }
    }

    bool Server::put(std::string key, std::string val) {
        pWBuffer->put_without_log(key, val);
        check_size();
        return true;
    }

    std::string Server::query(const std::string& key) {
        std::string res;
        for (int i = 0; i < WBufferNum; ++i) {
            int ind = (curWBufferInd + WBufferNum - i) % WBufferNum;
            WBufferCluster[ind]->query(key, res);
            if (!res.empty()) {
                break;
            }
        }
        if (res.empty()) {
            //TODO read_buffer
            bool flag = true;
            std::string_view buffer;
            while (flag){
                std::string val;
                lsmTree->query(key, val);
                if (val.empty())
                    return "";
                auto &recordVal = *reinterpret_cast<lsm::LsmRecordVal *>(const_cast<char *>(val.data()));
                buffer = scan_file(recordVal.fileNo, recordVal.recordOffset, recordVal.recordLen);
                if (!buffer.empty()) //如果buffer是空，说明正要scan时，gc删掉文件了没找到vFile，此时重新query； 概率极小，需要在gc写完新文件之后，release_snapshot之前查lsm，在remove之后查vFile
                    flag = false;
            }
            meta::f_off keyLen;
            std::memcpy(&keyLen, buffer.data() + sizeof(meta::f_off), sizeof(meta::f_off));
            auto ret_sv = buffer.substr(2*sizeof(meta::f_off)+keyLen, buffer.size()-2*sizeof(meta::f_off)-keyLen);
            res = ret_sv;
            delete buffer.data();
        }
        return res;
    }

    std::vector<std::pair<std::string, std::string >> Server::range_query(const std::string& startKey, const std::string& endKey, int len){
        meta::mtx4range.lock();
        using std::vector, std::pair, std::string, std::string_view;

        rangeQn ++;
        using namespace std::chrono;
        auto t1 = high_resolution_clock::now();

//        vector<pair<string_view,string_view>> rangeWBufferRes;
//        auto future = std::async(std::launch::async, [this, startKey, endKey, len, &rangeWBufferRes]() {
//            pWBuffer->range_query(startKey, endKey, len, rangeWBufferRes);
//        });

        auto t2 = high_resolution_clock::now();
        duration<double, std::micro> eee = t2 - t1;
        tm12 += eee.count();

        vector<pair<string , lsm::LsmRecordVal>> valLocations; //sorted
        auto static scanIndex = array<lsm::ScanInfo*, FILE_NO_MAX>{};
        lsmTree->range_query(startKey, endKey, len, valLocations, scanIndex);

        auto t3 = high_resolution_clock::now();
        eee = t3-t2;
        tm23 += eee.count();

        std::vector<std::pair<std::string_view, std::string_view >> rangeVStoreRes;
        find_vstore_kv(valLocations, scanIndex, rangeVStoreRes);

        auto t4 = high_resolution_clock::now();
        eee = t4 - t3;
        tm34 += eee.count();

//        future.get(); //在这里等
        vector<pair<string ,string >> result;
//        result.reserve(len);
//        merge_range(rangeWBufferRes, rangeVStoreRes, len, result);

//        for(auto & kvp : rangeVStoreRes){
//            result.emplace_back(kvp.first, kvp.second);
//        }

        auto t5 = high_resolution_clock::now();
        eee = t5-t4;
        tm45 += eee.count();

        auto maxFid = meta::readyFileNo.load();
        for (int i = 0; i < maxFid; ++i) {
            auto &buffer = meta::buffers[i];
            delete[] buffer.data();
            buffer = {nullptr, 0};

            auto &shouldScan = scanIndex[i];
            delete shouldScan;
            shouldScan = nullptr;
        }

        meta::mtx4range.unlock();
        return result;
    }

    std::vector<std::pair<std::string, std::string >> Server::range_query2(const std::string& startKey, const std::string& endKey, int len){
        meta::mtx4range.lock();

        rangeQn ++;
        using namespace std::chrono;
        auto t1 = high_resolution_clock::now();


        vector<pair<string_view, string_view>> rangeWBufferRes;
        auto future = std::async(std::launch::async, [this, startKey, endKey, len, &rangeWBufferRes]() {
            pWBuffer->range_query(startKey, endKey, len, rangeWBufferRes);
        });

        auto t2 = high_resolution_clock::now();
        duration<double, std::micro> eee = t2 - t1;
        tm12 += eee.count();

        std::vector<std::pair<std::string , std::string>> res; //sorted
        lsmTree->range_query2(startKey, endKey, len, res);

        auto t3 = high_resolution_clock::now();
        eee = t3-t2;
        tm23 += eee.count();

        auto t4 = high_resolution_clock::now();

        future.get(); //在这里等

        auto t5 = high_resolution_clock::now();
        eee = t5-t4;
        tm45 += eee.count();

        meta::mtx4range.unlock();
        return res;
    }

    void Server::init() {
        basic_init();
        lsm_init();
        WBuffer_init();
//        daemon_init();
    }

    void Server::persist_buffer(vstore::WBuffer *snapshotWBuffer) {
        snapshotWBuffer->dump();
    }

    void Server::change_ind() {
        pWBuffer->status = 1;
        curWBufferInd = (curWBufferInd + 1) % WBufferNum;
        pWBuffer = WBufferCluster[curWBufferInd];
        while (pWBuffer->status.load() == 1) ; //还没有写完磁盘，不能重置此buffer，写完之后会变成0
        pWBuffer->ready();
    }

    std::string_view Server::scan_file(meta::fid fileNo, meta::f_off startOffset, size_t scanLen){
        if(meta::fileControlInfo.find(fileNo) == meta::fileControlInfo.end())
//            return "";
            return {nullptr, 0};

        meta::fileControlInfo[fileNo].readers.fetch_add(1);

//        auto vFilePathStr = meta::fileControlInfo[fileNo].addr;
//        int fd = open(vFilePathStr.c_str(), O_RDONLY);
//        if (fd == -1) {
//            perror("Error opening vFile");
//        }
        int fd = fileno(meta::fileControlInfo[fileNo].fd);
        // Set the file offset
        if (lseek(fd, startOffset, SEEK_SET) == -1) {
            perror("Error setting file offset");
            close(fd);
        }

        // Read data into buffer
        char *buffer = new char[scanLen];

        auto bytesRead = read(fd, buffer, scanLen);
        if (bytesRead == -1) {
            perror("Error reading file");
            free(buffer);
//            close(fd);
            meta::fileControlInfo[fileNo].readers.fetch_sub(1);
            return {nullptr, 0};
        }
//        close(fd);

        meta::fileControlInfo[fileNo].readers.fetch_sub(1);

        return {buffer, scanLen};
    }

    void Server::basic_init() {
        WBufferNum = conf::data.value_store.buffer_num;
        curWBufferInd = 0;
        threshold = conf::data.value_store.buffer_size;
        fs::create_directories(conf::data.db.vstore_path);
        fs::create_directories(conf::data.db.lsm_path);
        fs::create_directories(conf::data.db.meta_path);
    }

    void Server::lsm_init() {
        lsmTree = new lsm::LevelDB(conf::data.db.lsm_path);
        bloom_parameters parameters;
        // How many elements roughly do we expect to insert?
        parameters.projected_element_count = 32*1024*1024;
        // Maximum tolerable false positive probability? (0,1)
        parameters.false_positive_probability = 0.001; // 1 in 1000
        // Simple randomizer (optional)
        parameters.random_seed = 0xA5A5A5A5;
        if (!parameters)
        {
            std::cout << "Error - Invalid set of bloom filter parameters!" << std::endl;
        }
        parameters.compute_optimal_parameters();

        //Instantiate Bloom Filter
        LSMTreeBloomFiler = bloom_filter{parameters};
    }

    void Server::WBuffer_init() {
        auto vstorePath = conf::data.db.vstore_path;
        for (int i = 0; i < WBufferNum; ++i) {
            WBufferCluster.push_back(new vstore::OrderkvWBuffer());
        }
        pWBuffer = WBufferCluster[0];
//        bufferImage_load(pWBuffer);
    }

    void Server::daemon_init() {
        std::thread daemonThread([&]() {
            while (!meta::closeDB.load()) {
                // 等待1秒
                std::this_thread::sleep_for(std::chrono::seconds(1));
                // 调用垃圾监测函数
                garbage_monitoring();
            }
        });
        daemonThread.detach();
    }

    void Server::check_size() {
        if (pWBuffer->size >= threshold) {
            vstore::WBuffer *snapshotWBuffer = pWBuffer;
            change_ind();
            persistence(snapshotWBuffer);
        }
    }

    void Server::persistence(vstore::WBuffer *snapshotWBuffer) {
        std::thread( [this, snapshotWBuffer](){
            meta::mtx.lock();
//            auto ret1 = std::async(std::launch::async, [this, snapshotWBuffer](){garbage_compute(snapshotWBuffer);} );
            garbage_compute(snapshotWBuffer);
            auto ret2 = std::async(std::launch::async, [snapshotWBuffer](){persist_buffer(snapshotWBuffer);} );
            auto ret3 = std::async(std::launch::async, [this, snapshotWBuffer](){write_lsm(dynamic_cast<vstore::OrderkvWBuffer *>(snapshotWBuffer));} );
//            ret1.get();
            ret2.get();
            ret3.get();
            fs::remove(snapshotWBuffer->loggerFilePathStr); //如果这个logFile还存在，说明发生了crash，重启时需要把数据从logFile重新load进buffer，此时会重新触发write。如果已经删除，则代表此次持久化已经成功。
            snapshotWBuffer->status = 0;
            meta::mtx.unlock();
            garbage_monitoring();
        }).detach();
    }

    void Server::write_lsm(vstore::OrderkvWBuffer *snapshotWBuffer) {
//        lsmTree->snapshot();
        auto& input = snapshotWBuffer->buffer;
        meta::f_off offset = 0;
        /*for (auto &lenRecordPair: input) {
            auto& key = lenRecordPair.first;
            auto record = &lenRecordPair.second.second;
            record->recordOffset = offset;
            offset += (meta::f_off )record->recordLen;
            auto recordCast = reinterpret_cast<char *>(record);
            lsmTree->put(key, recordCast);
        }*/
        int ind = 0;
        std::vector<std::pair<std::string_view, std::string_view>> kvPairBatch;
        kvPairBatch.reserve(BATCH_SIZE);

        for (auto &lenRecordPair: input) {
            auto &key = lenRecordPair.first;
            LSMTreeBloomFiler.insert(key);
            auto record = &lenRecordPair.second.second;
            record->recordOffset = offset;
            offset += (meta::f_off) record->recordLen;
            auto recordCast = reinterpret_cast<char *>(record);

            kvPairBatch.emplace_back(key, recordCast);

            ++ind;
            if (ind == BATCH_SIZE) {
                lsmTree->batch_put(kvPairBatch);
                kvPairBatch.clear();
                ind = 0;
            }
        }
        if (!kvPairBatch.empty()) {
            lsmTree->batch_put(kvPairBatch);
            kvPairBatch.clear();
        }
//        lsmTree->release_snapshot();
    }

    inline void Server::find_vstore_kv(vector<pair<string , lsm::LsmRecordVal>>& valLocations,
                                                           array<lsm::ScanInfo*, FILE_NO_MAX>& scanIndex,
                                                           vector<pair<string_view, string_view >>& ret){

        using namespace std;
        using namespace std::chrono;
        ret.reserve(valLocations.size());
        auto t1 = high_resolution_clock::now();

        auto t2 = high_resolution_clock::now();
        duration<double, std::micro> eee = t2 - t1;
        scanTm12 += eee.count();

        static ThreadPool threadPool(conf::data.db.scan_thread_num);
//        map<meta::fid, shared_future<string_view>> buffers;
        array<shared_future<string_view>, FILE_NO_MAX> buffersFuture{};
        unsigned int maxFid = meta::readyFileNo.load();
        for(int fid = 0; fid < maxFid; ++fid){
            if (scanIndex[fid] == nullptr)
                continue;
            auto shouldScan = *(scanIndex[fid]);
            auto buffer_future = threadPool.enqueue(scan_file, fid, shouldScan.startOff, shouldScan.endOff - shouldScan.startOff);
            buffersFuture[fid] = std::move(buffer_future.share());
        }

//        map<meta::fid, string_view> buffers;
//        map<meta::fid, bool> oks{};
//        for(int fid = 0; fid < scanIndex.size(); ++fid){
//            if (scanIndex[fid] == nullptr)
//                continue;
//            auto shouldScan = *(scanIndex[fid]);
//            auto buffer = scan_file(fid, shouldScan.startOff, shouldScan.endOff - shouldScan.startOff);
//            _scanthreads.schedule(
//                    std::bind(
//                            scan_file2,
//                            fid,
//                            shouldScan.startOff,
//                            shouldScan.endOff - shouldScan.startOff,
//                            boost::ref(buffers[fid]),
//                            boost::ref(oks[fid])
//                    )
//            );
//        }

//        map<meta::fid, string_view> buffers;
//        for(int fid = 0; fid < scanIndex.size(); ++fid){
//            if (scanIndex[fid] == nullptr)
//                continue;
//            auto shouldScan = *(scanIndex[fid]);
//            auto buffer = scan_file(fid, shouldScan.startOff, shouldScan.endOff - shouldScan.startOff);
//            buffers.emplace(fid, buffer);
//        }

        auto t3 = high_resolution_clock::now();
        eee = t3 - t2;
        scanTm23 += eee.count();

        for(auto& kvp : valLocations){
            auto& key = kvp.first;
            auto& location = kvp.second;
            auto fid = location.fileNo;
            if (meta::buffers[fid].empty()){
                meta::buffers[fid] = buffersFuture[fid].get();
            }
            auto buffer = meta::buffers[fid];
            if (buffer.empty())
                continue;
            meta::f_off startScanOffsetOfFile = (*scanIndex[location.fileNo]).startOff;
            meta::f_off startScanOffsetOfBuffer_record = location.recordOffset - startScanOffsetOfFile;
//            meta::f_off startScanOffsetOfBuffer_keyLenInfo = startScanOffsetOfBuffer_record + sizeof(meta::f_off);
//            meta::f_off keyLen;
//            std::memcpy(&keyLen, buffer.data()+startScanOffsetOfBuffer_keyLenInfo, sizeof(meta::f_off));
//            meta::f_off scanLen_value = location.recordLen - 2*sizeof(meta::f_off) - keyLen;
//            meta::f_off startScanOffsetOfBuffer_value = startScanOffsetOfBuffer_record + 2*sizeof(meta::f_off) + keyLen;
//            auto valStr = buffer.substr(startScanOffsetOfBuffer_value, scanLen_value);
            auto valStr = buffer.substr(startScanOffsetOfBuffer_record, location.recordLen);
            ret.emplace_back(kvp.first, valStr);
        }

        auto t4 = high_resolution_clock::now();
        eee = t4 - t3;
        scanTm34 += eee.count();

        auto t5 = high_resolution_clock::now();
        eee = t5 - t4;
        scanTm45 += eee.count();
   }

    inline void Server::merge_range(vector<pair<string_view, string_view>>& wBufferRes,
                                    vector<pair<string_view, string_view>>& vstoreRes,
                                    int len,
                                    std::vector<std::pair<std::string, std::string >>& ret){
        ret.reserve(len);
        auto wit = wBufferRes.begin();
        auto vit = vstoreRes.begin();
        int resultNum = 0;
        while (wit != wBufferRes.end() && vit != vstoreRes.end() && resultNum < len) {
            auto cmp = wit->first.compare(vit->first);
            if (cmp < 0) {
                ret.emplace_back(*wit);
                ++wit;
            } else if (cmp > 0) {
                ret.emplace_back(*vit);
                ++vit;
            } else {
                ret.emplace_back(*wit);
                ++wit;
                ++vit;
            }
            ++resultNum;
        }
        while (wit != wBufferRes.end() && resultNum < len) {
            ret.emplace_back(*wit);
            ++wit;
            ++resultNum;
        }
        while (vit != vstoreRes.end() && resultNum < len) {
            ret.emplace_back(*vit);
            ++vit;
            ++resultNum;
        }
    }

    void Server::garbage_compute(vstore::WBuffer *wBuffer) {
        auto& data = ((vstore::OrderkvWBuffer *) wBuffer)->buffer;
        std::string garbageLoggerPathStr = conf::data.db.meta_path + "/garbageLog" + ((vstore::OrderkvWBuffer *) wBuffer)->fileNo8Digits + ".txt";
//        std::ofstream garbageLogger(garbageLoggerPathStr);
        for (const auto &item: data){
            auto key = item.first;
            auto contain = LSMTreeBloomFiler.contains(key);
            if (!contain)
                continue;
            std::string res;
            lsmTree->query(key, res);
            if (res.empty())
                continue;
            auto &recordVal = *reinterpret_cast<lsm::LsmRecordVal *>(const_cast<char *>(res.data()));
//                garbageLogger << recordVal;
            auto fid = recordVal.fileNo;
            meta::fileControlInfo[fid].garbageSize += recordVal.recordLen;
            meta::fileControlInfo[fid].invalidValOffsets.emplace_back(recordVal.recordOffset);
        }
//        garbageLogger << "finished";
    }

    void Server::garbage_monitoring(){
        meta::fid maxGarbageFid = 0;
        size_t maxGarbageSize = 0;
        for (auto &it: meta::fileControlInfo) {
            if (it.second.garbageSize > maxGarbageSize) {
                maxGarbageSize = it.second.garbageSize;
                maxGarbageFid = it.first;
            }
        }
        if (maxGarbageSize > 0){ //TODO env改成变量吧
            meta::mtx.lock();
            meta::mtx4range.lock();
            gc(maxGarbageFid);
            meta::mtx4range.unlock();
            meta::mtx.unlock();
//            meta::storeInfo();
        }
    }

    void Server::gc(meta::fid fid){
        auto path_gc = meta::fileControlInfo[fid].addr;
//        int fd_gc = open(path_gc.c_str(), O_RDONLY);
        int fd_gc = fileno(meta::fileControlInfo[fid].fd);
        if (fd_gc == -1) {
            perror(("Error opening " + std::to_string(fid) + " for reading").c_str());
            return;
        }
        struct stat fileStat{};
        if (fstat(fd_gc, &fileStat) == -1) {
            perror("fstat! OMG!");
            return;
        }
        auto fileSize = fileStat.st_size;
//        off_t fileSize = lseek(fd_gc, 0, SEEK_END);
        lseek(fd_gc, 0, SEEK_SET);

        // 映射文件A到内存，先读后写的情况下，mmap比read略快
        auto* mmapData = (char *)mmap(nullptr, fileSize, PROT_READ, MAP_PRIVATE, fd_gc, 0);

        if (mmapData == MAP_FAILED) {
            //TODO log
            perror("Error mapping file FD_GC to memory");
            return;
        }

//        meta::obsoleteFidLogger << "readyToDel:" << std::endl;
//        meta::obsoleteFidLogger << fid << std::endl;

        auto cleanFileNo = ++meta::readyFileNo;
//        meta::obsoleteFidLogger << "cleanFileNo" << std::endl;
//        meta::obsoleteFidLogger << cleanFileNo << std::endl;
//        meta::obsoleteFidLogger << "ok" << std::endl;

        auto vstorePath = conf::data.db.vstore_path;
        std::ostringstream ss;
        ss << std::setw(8) << std::setfill('0') << cleanFileNo;
        auto cleanFilePathStr = vstorePath + "/vFile" + ss.str() + ".data";
//        std::ofstream outFile(cleanFilePathStr);
        FILE *outputF = fopen(cleanFilePathStr.c_str(), "w+");

        auto offs = meta::fileControlInfo[fid].invalidValOffsets;
        std::sort(offs.begin(), offs.end());
        meta::f_off curOff = 0;
        auto offIter = offs.begin();
        meta::f_off cleanFileOff = 0;
        lsmTree->snapshot();
        while (curOff < fileSize) {
            meta::f_off recordLen;
            char* curAddr = mmapData + curOff;
            std::memcpy(&recordLen, curAddr, sizeof(meta::f_off));
            if (curOff == *offIter){ //忽略此行
                ++offIter;
            } else {
                meta::f_off keyLen;
                std::memcpy(&keyLen, curAddr + sizeof(meta::f_off), sizeof(meta::f_off));
                std::string key = {curAddr+2*sizeof(meta::f_off), keyLen};
                lsm::LsmRecordVal record = {cleanFileNo, cleanFileOff, recordLen};
                lsmTree->put(key, {reinterpret_cast<char *>(&record), sizeof(lsm::LsmRecordVal)});
//                outFile.write(curAddr, recordLen);
                fwrite(curAddr, sizeof(char), recordLen, outputF);
                cleanFileOff += recordLen;
            }
            curOff += recordLen;
        }
//        outFile.flush();
        fflush(outputF);
//        outFile.close();
        auto cleanFileSize = fs::file_size(cleanFilePathStr);
        if (cleanFileSize == 0) {
            fs::remove(cleanFilePathStr);
        }else{
//            meta::fidAddrLogger << cleanFileNo << " " << cleanFilePathStr << std::endl;
            meta::fileControlInfo[cleanFileNo].addr = std::move(cleanFilePathStr);
            meta::fileControlInfo[cleanFileNo].fd = outputF;
//            meta::fidAddrLogger << "ok" << std::endl;
        }

        lsmTree->release_snapshot();

        if (munmap(mmapData, fileSize) == -1) {    // 解除内存映射
            //TODO log perror("Error unmapping memory");
            return;
        }

        //等待reader归零，此时没有索引此vFile，此时可以删除老文件
        while (meta::fileControlInfo[fid].readers.load());

//        meta::obsoleteFidLogger << fid << std::endl;

        fclose(meta::fileControlInfo[fid].fd);
        //移除老文件
        remove(path_gc.c_str());

        //删除fid
        meta::fileControlInfo.erase(fid);

        //right sequence example:    readyToGC 192 cleanFileNo 786 192
        //修复： 如果正确，则clean一下192就好，如果错误，是否有786，如果有，则clean786，如果没有786，则重新gc192
    }

    void Server::bufferImage_store(vstore::WBuffer *wBuffer) {
        using namespace std;
        auto& data = ((vstore::OrderkvWBuffer *) wBuffer)->buffer;
        string bufferImagePathStr = conf::data.db.vstore_path + "/buffer_image.txt";
        filesystem::path bufferImagePath(bufferImagePathStr);
        create_directories(bufferImagePath.parent_path());
        ofstream output(bufferImagePath);
        if (data.empty()) {
            output.close();
            return;
        }
        output << data.size() << endl << data.begin()->second.second.fileNo << endl;
        for (auto &kvPair: data) {
            auto &key = kvPair.first;
            auto &val = kvPair.second.first;
            output << key << endl << val << endl;
        }
        output.close();
    }

    void Server::bufferImage_load(vstore::WBuffer *wBuffer) {
        using namespace std;
        string bufferImagePathStr = conf::data.db.vstore_path + "/buffer_image.txt";
        filesystem::path bufferImagePath(bufferImagePathStr);
        ifstream input(bufferImagePath);
        if (!input.is_open())
            return;
        int dataSize;
        input >> dataSize >> ((vstore::OrderkvWBuffer *) wBuffer)->fileNo;
        for (int i = 0; i < dataSize; ++i) {
            string key, val;
            input >> key >> val;
            ((vstore::OrderkvWBuffer *) wBuffer)->put(key, val);
        }
        input.close();
    }

    void Server::crash_check_gc() {
        using namespace std;
        ifstream input(meta::obsoleteFidLoggerPathStr);
        if (!input.is_open())
            return; //不存在此文件，跳过
        string readyFidStr, cleanFidStr,confirmFidStr, tmp, OK; //readyToGC 192 cleanFileNo 786 192
        while (input >> tmp >> readyFidStr >> tmp >> cleanFidStr >> OK >> confirmFidStr);
        if(!readyFidStr.empty() && readyFidStr == confirmFidStr)
            return; //gc完整，不管
        if (OK == "ok") {
            meta::fid cleanFileNo = std::stoi(cleanFidStr);
            std::ostringstream ss;
            ss << std::setw(8) << std::setfill('0') << cleanFileNo;
            auto cleanFilePathStr = conf::data.db.vstore_path + "/vFile" + ss.str() + ".data";
            std::filesystem::remove(cleanFilePathStr);
        }
    }

    void Server::crash_check_data(){
        //log1 log2 log3
        //肯定只有最后一个没满，log3将设置logger（按照fileNo而不是ready）
        using namespace std;
        auto vstorePath = conf::data.db.vstore_path;
        vector<string> logFilePathStrVec;
        for (const auto& entry : fs::directory_iterator(vstorePath)) {
            if (entry.is_regular_file() && entry.path().filename().string().find("log") == 0) {
                logFilePathStrVec.emplace_back(entry.path());
            }
        }
        if (logFilePathStrVec.empty()) {
            pWBuffer->ready();
            return;
        }
        sort(logFilePathStrVec.begin(), logFilePathStrVec.end());

        for (int i = 0; i < logFilePathStrVec.size(); ++i) {
            ifstream log(logFilePathStrVec[i]);
            log >> WBufferCluster[i]->fileNo;
            string key, val;
            while (log >> key >> val) {
                WBufferCluster[i]->put_without_log(key, val);
            }
            if (i != logFilePathStrVec.size() - 1) { //不是最后的log，直接persist
                persistence(WBufferCluster[i]);
            } else {
                std::ostringstream ss;
                ss << std::setw(8) << std::setfill('0') << WBufferCluster[i]->fileNo;
                WBufferCluster[i]->fileNo8Digits = ss.str();
                // 文件路径
                WBufferCluster[i]->loggerFilePathStr = vstorePath + "/log" + WBufferCluster[i]->fileNo8Digits + ".txt";
                std::filesystem::path filePath = WBufferCluster[i]->loggerFilePathStr;
                //创建文件夹
                std::filesystem::create_directories(filePath.parent_path());
                WBufferCluster[i]->logger = fopen(WBufferCluster[i]->loggerFilePathStr.data(), "a");

                pWBuffer = WBufferCluster[i];
            }
        }
    }
}